package ds_industry_2025.ds.ds_06.tzgc_tjxt

import org.apache.spark.mllib.linalg.{DenseVector, Vector, Vectors}
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

import scala.collection.mutable.ArrayBuffer

/*
    1、根据子任务一的结果，对其进行SVD分解，对数据进行降维保留前5个奇异值信息，根据该用户已购买的商品分别与未购买的商品计算余弦相
    似度再进行累加求均值，将均值最大的5件商品id进行输出作为推荐使用。将输出结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】
    中对应的任务序号下。
结果格式如下：

------------------------推荐Top5结果如下------------------------
相似度top1(商品id：1，平均相似度：0.983456)
相似度top2(商品id：71，平均相似度：0.782672)
相似度top3(商品id：22，平均相似度：0.7635246)
相似度top4(商品id：351，平均相似度：0.7335748)
相似度top5(商品id：14，平均相似度：0.522356)
 */
object t3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("tj")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    //  todo 读取特征工程第二题的结果
    val source=spark.table("tzgc.skumat")

    //  todo 拿到0.0用户的所有购买信息
    val skus:Seq[Double] = source.filter(col("user_id") === 0.0).limit(1)
      .select(source.columns.toSeq.slice(1, source.columns.toSeq.length).map(col): _*)
      .map(r => r.toSeq.map(r => r.toString.toDouble))
      .collect()(0)

    //  todo 创建两个数组存储用户0.0的购买情况
    var buy=new ArrayBuffer[Double]()
    var no_buy=new ArrayBuffer[Double]()

    skus.zipWithIndex.foreach(
      e =>{
        if(e._1.equals(0.0)){
          no_buy += e._2
        }else{
          buy += e._2
        }
      }
    )

    //  todo 将除了id字段外所有的字段的数据转化为向量，为了方便后面转化为RowMatrix
    //  todo 特别注意这里需要使用rdd,因为vector是一个非规则化的数据
    val mapData = source.rdd.map(
      r => {
        val v: Vector = Vectors.dense(r.toSeq.slice(1, r.toSeq.length).map(r => r.toString.toDouble).toArray)
        v
      }
    ).collect()

    //  todo 将数据转化为RowMatrix
    val matrix = new RowMatrix(spark.sparkContext.parallelize(mapData))

    //  todo 对数据进行svd分解
    val svd = matrix.computeSVD(5, computeU = true)

    //  todo 拿到右奇异值
    val v = svd.V

    //  todo 给每一行添加一个索引，为了后面转化为dataframe
    val to_array:Array[(Vector,Int)] = v.rowIter.zipWithIndex.toArray

    to_array.foreach(println)

    //  todo 将数据转化为dataframe
    val sku_info = spark.createDataFrame(
      to_array
    ).toDF("vec", "id")

    //  todo 注册余弦相似度的函数
    spark.udf.register(
      "cos",
      (v1:DenseVector,v2:DenseVector) => {
        1 - breeze.linalg.functions.cosineDistance(
          breeze.linalg.DenseVector(v1.values),
          breeze.linalg.DenseVector(v2.values)
        )
      }
    )

    //  todo 交叉连接得到结果
    val result = sku_info.crossJoin(sku_info)
      .toDF("vec", "id", "vec2", "id2")
      .filter(col("id") !== col("id2"))
      .withColumn("cos", expr("cos(vec,vec2)"))
      .filter(col("id").isin(buy: _*) && col("id2").isin(no_buy: _*))
      .groupBy("id2")
      .agg(avg("cos").as("avg_cos"))
      .orderBy(desc("avg_cos"))

    println("----------推荐Top5结果如下---------------")
    result.limit(5).collect().zipWithIndex.foreach{
      case(r,index) =>
        val id=r.getAs[Int]("id2")
        val cos=r.getAs[Double]("avg_cos")
        val str=s"相似度top${index+1}(商品id:${id},平均相似度:${cos})"
        println(str)

    }








    spark.close()
  }

}
